package org.yuanzheng.time

import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import org.yuanzheng.source.StationLog

/**
 * @author yuanzheng
 * @date 2020/6/23-21:01
 */
object LatenessDataOnWindow {
  def main(args: Array[String]): Unit = {
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    import org.apache.flink.streaming.api.scala._

    //设置时间语义
    streamEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    //streamEnv.getConfig.setAutoWatermarkInterval(100L) //周期引入watermark的设置，默认是100毫秒

    //读取数据源
    val stream: DataStream[StationLog] = streamEnv.socketTextStream("192.168.1.10", 8888)
      .map(line => {
        var split = line.split(",")
        new StationLog(split(0).trim, split(1).trim, split(2).trim, split(3).trim, split(4).trim.toLong, split(5).trim.toLong)
      })

      //引入watermark，数据是乱序的，并且大多数数据延迟2秒
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[StationLog](Time.seconds(2)) { //水位线延迟2秒
        override def extractTimestamp(element: StationLog): Long = {
          element.callTime
        }
      })

    val lateTag = new OutputTag[StationLog]("late")
    //分组、开窗
    val result: DataStream[String] = stream.keyBy(_.sid).timeWindow(Time.seconds(10), Time.seconds(5))
      //数据延迟超出2秒，allowedLateness处理
      //第一种情况：允许数据迟到5秒（迟到2-5秒）再次触发窗口函数。触发条件时 watermark < endOfWindow + allowedLateness
      .allowedLateness(Time.seconds(5))
      //第二种情况：迟到数据5秒以上，输出到侧流
      .sideOutputLateData(lateTag)
      //聚合函数
      .aggregate(new MyAggregateCountFunction, new OutputResultWindowFunction)

    //打印数据
    result.getSideOutput(lateTag).print("late")
    result.print("main")
    streamEnv.execute()
  }
}

class MyAggregateCountFunction extends AggregateFunction[StationLog, Long, Long] {
  override def createAccumulator(): Long = 0

  override def add(in: StationLog, acc: Long): Long = acc + 1

  override def getResult(acc: Long): Long = acc

  override def merge(acc: Long, acc1: Long): Long = acc + acc1
}

class OutputResultWindowFunction extends WindowFunction[Long, String, String, TimeWindow] {
  override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[String]): Unit = {
    var value = input.iterator.next()
    var sb = new StringBuilder
    sb.append("窗口范围是:").append(window.getStart).append("---").append(window.getEnd)
      .append("\n")
      .append("当前基站ID是:").append(key)
      .append("，呼叫的数据量是:").append(value)
    out.collect(sb.toString())
  }
}